package com.zhl.hadoop.cases.phone;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * @program: demos
 * @description:
 * @author: 刘振华
 * @create: 2020-10-28 11:22
 **/
public class PhoneFlowJobMain extends Configured implements Tool {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

//		conf.set("fs.defaultFS", "hdfs://zhlcluster");
		//启动job任务 0表示成功
		int run = ToolRunner.run(conf,new PhoneFlowJobMain(), args);
		System.exit(run);
	}
	//指定一个job任务，配置工作  /zhltest	/output
	@Override
	public int run(String[] args) throws Exception {
		//1.创建一个job对象
		Job job = Job.getInstance(super.getConf(), "phoneFlow");

		//报找不到mapper  class的错误，就添加这句
		job.setJarByClass(PhoneFlowJobMain.class);

		//2.配置job的8个步骤，可作为模板
		//第一步：指定文件读取方式和读取路径
		job.setInputFormatClass(TextInputFormat.class);
		TextInputFormat.addInputPath(job,new Path(args[0]));
//		TextInputFormat.addInputPath(job,new Path("file:///D:\\课程\\2.大数据\\测试数据"));

		//第二步：指定map阶段的代码处理逻辑
		job.setMapperClass(PhoneFlowMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(PhoneFlow.class);

		//3，4，5，6采用默认方式处理

//		job.setPartitionerClass(FlowPartitioner.class);
		job.setCombinerClass(PhoneFlowCombiner.class);

		//第七步：自定义reduce逻辑
		job.setReducerClass(PhoneFlowReduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

//		job.setNumReduceTasks(3);

		//第八步：输出类型和路径
		job.setOutputFormatClass(TextOutputFormat.class);
		Path outPath = new Path(args[1]);
//		FileSystem fs = FileSystem.get(super.getConf());
//		if(fs.exists(outPath)){
//			fs.delete(outPath,true);
//		}
		TextOutputFormat.setOutputPath(job,outPath);
//		TextOutputFormat.setOutputPath(job,new Path("file:///D:\\课程\\2.大数据\\output"));

		boolean rs = job.waitForCompletion(true);
		return rs?0:1;
	}
}
